package day03;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.net.URI;
import java.util.Scanner;

/**
 * 进行状态管理 实现CheckpointedFunction
 */
public class RichHdfsSourceFunction extends RichSourceFunction<String> implements CheckpointedFunction {

    Boolean isRunning=true;
    private String url=null;
    //每次读取文件保存最后状态的md5值 初始md5不能为空 加入不了状态
    private String md5="";
    //每隔 时间扫描
    private static final int interval=3000;
    Path path=null;
    FileSystem fs=null;

    public RichHdfsSourceFunction(String url) {
        //初始化执行的代码在driver端
        this.url =(url);
    }

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        Configuration conf = new Configuration();
        fs=FileSystem.get(new URI("hdfs://ns1"),conf,"root");
        path = new Path(this.url);
        System.out.println("-->open");
    }

    @Override
    public void close() throws Exception {
        System.out.println("-->close");
        super.close();
    }

    //hdfs连接的时候需要指定配置文件 core-site.xml   hdfs-site.xml
    @Override //executor端
    public void run(SourceContext<String> ctx) throws Exception {
        while(isRunning){
            //如果不存在跳过当次
            if(!fs.exists(path)){
                Thread.sleep(interval);
                continue;
            }

            //获取校验核文件
            String fileChecksum = fs.getFileChecksum(path).toString();
            //获取文件md5()
            String current_md5 = fileChecksum.substring(fileChecksum.indexOf(":") + 1);

            if(this.md5==null||!current_md5.equals(this.md5)){
                //存在就读
                FSDataInputStream in = fs.open(path);
                Scanner sc= new Scanner(in);
                while (sc.hasNext()){
                    String line = sc.nextLine();
                    ctx.collect(line);
                }
                //更新md5
                this.md5=current_md5;
                //关闭流
                in.close();
            }
            Thread.sleep(interval);

        }
    }

    @Override
    public void cancel() {
        System.out.println("-->cancel");
        isRunning=false;
    }

    private ListState<String> listState;

    //将数据放到状态
    @Override//1s一次
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        System.out.println("snapshotState");
        listState.add(this.md5);
    }

    // 第一次启动    初始化状态
    // 不是第一次启动  恢复状态
    @Override//每次启动一次
    public void initializeState(FunctionInitializationContext context) throws Exception {
        System.out.println("initializeState");

        //不管是初始化也好还是恢复数据也好都是一个方法
        ListStateDescriptor<String> desc = new ListStateDescriptor<>("listState", String.class);//描述器
        this.listState =context.getOperatorStateStore().getListState(desc);
        //判断context.isRestored()为恢复状态 将md5恢复 单线程用next取
        if(context.isRestored()){
            this.md5=this.listState.get().iterator().next();
        }
        System.out.println(this.md5);

    }
}
